package com.atguigu.realtime.app.dim;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.func.MyBroadcastProcessFunction;
import com.atguigu.realtime.app.func.MyPhoenixSink;
import com.atguigu.realtime.bean.CDCJavaBean;
import com.atguigu.realtime.utils.KafkaUtil;
import com.sun.xml.internal.bind.v2.TODO;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.ZoneId;

public class dimSinkApp {
    public static void main(String[] args) throws Exception {
        //1、TODO 获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        //设置时区
        tableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));
        //超时时间
//        tableEnvironment.getConfig().setIdleStateRetention(Duration.ofSeconds(905L));

        //2.TODO 设置状态后端
        //设置五分钟开启一次checkpoint，模式设置为精准一次（默认）
//        //也可以这么写（分开写）
////        env.enableCheckpointing(5 * 60 * 1000L);
////        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
//        //设置checkpoint超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
//        //设置出现最多个检查点的数量（有可能第一个检查点没结束，第二个就开始了）
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        //配置状态后端(FsStateBack的状态后端设置)
//        env.setStateBackend(new HashMapStateBackend());
//        //写入hdfs需要设置用户
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
//        //写明状态后端保存的地址
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8082/gmall/ck");

        //3.TODO 读取topic_db数据
        String topicName = "topic_db";
        String groupID = "dim_sink_app";
        DataStreamSource<String> dbStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));
        //4.TODO 转换为JSON文件  过滤脏数据  写入侧输出流
        //指定侧输出流
        OutputTag<String> dirtyStram = new OutputTag<String>("dirty"){};
        SingleOutputStreamOperator<JSONObject> processStream = dbStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    String type = jsonObject.getString("type");
                    if (!"bootstrap-start".equals(type) && !"bootstrap-complete".equals(type)) {
                        out.collect(jsonObject);
                    } else {
                        System.out.println("staet or complete 数据写入侧输出流");
                        ctx.output(dirtyStram, value);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("JSON转换失败，数据写出到dirtyStream");
                    ctx.output(dirtyStram, value);
                }
            }
        });
        //5.TODO 读取CDC配置表信息  配置为广播流
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("CDCTable")
                .tableList("CDCTable.table_process")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> cdcStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),
                "cdcTable");
        //6.TODO 广播状态
        MapStateDescriptor<String, CDCJavaBean> mapState = new MapStateDescriptor<>("table_process", String.class,
                CDCJavaBean.class);
        BroadcastStream<String> broadcastStream = cdcStream.broadcast(mapState);
        //7.TODO 连接主流和配置流
        BroadcastConnectedStream<JSONObject, String> connectStream = processStream.connect(broadcastStream);
        //8.TODO 处理连接流  根据配置流的信息  过滤出主流的维度表内容
        SingleOutputStreamOperator<JSONObject> filterStream =
                connectStream.process(new MyBroadcastProcessFunction(mapState));

        filterStream.print("filter>>>>>>>>>>>>");
        //8.TODO 将数据写入phoenix中
        filterStream.addSink(new MyPhoenixSink());
        //9.TODO 执行任务
        DataStream<String> sideOutput = processStream.getSideOutput(dirtyStram);
        sideOutput.print("脏数据:");
        env.execute(groupID);
    }
}
